package bjut.zrpc.core.optimization.utils;

import bjut.zrpc.core.optimization.constant.ZRpcConstant;
import bjut.zrpc.core.optimization.enums.ZRpcConfigEnum;
import bjut.zrpc.core.optimization.factory.SingletonFactory;
import bjut.zrpc.core.optimization.protect.ServicesRecord;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhuzhiming
 * @Date: 2022/6/23 21:43
 */
@Slf4j
public class CuratorUtil {

    private static CuratorFramework zkClient;
    private static final String DEFAULT_ZOOKEEPER_ADDRESS = "127.0.0.1:2181";
    private static final int SLEEP_TIME = 3000;
    private static final int RETRY = 3;

    private static final Set<String> REGISTERED_SERVICE_PATH_SET = ConcurrentHashMap.newKeySet();
    private static final ConcurrentHashMap<String, List<String>> SERVICES_ADDRESS_MAP = new ConcurrentHashMap<>();
    private static final ServicesRecord SERVICES_RECORD = SingletonFactory.getInstance(ServicesRecord.class);


    public static CuratorFramework getZkClient() {

        if (zkClient != null && zkClient.getState() == CuratorFrameworkState.STARTED) {
            return zkClient;
        }
        String[] fileNames = ZRpcConfigEnum.ZRPC_CONFIG_NAME.getPropertyValue();
        Properties properties;
        String zookeeperAddress = "";
        for (String fileName : fileNames) {
            properties = PropertiesReadUtil.readProperties(fileName);
            if (properties == null) {
                continue;
            }
            zookeeperAddress = properties.getProperty(ZRpcConfigEnum.ZOOKEEPER_SERVER_CONFIG.getPropertyValue()[0]);
            if (zookeeperAddress != null && !"".equals(zookeeperAddress)) {
                break;
            }
        }

//        if (properties == null )
//            throw new RuntimeException(ZRpcConfigEnum.ZRPC_CONFIG_NAME.getPropertyValue()[0] + " not exist");

        zookeeperAddress = (zookeeperAddress == null || zookeeperAddress.equals("")) ? DEFAULT_ZOOKEEPER_ADDRESS : zookeeperAddress;
        ExponentialBackoffRetry backoffRetry = new ExponentialBackoffRetry(SLEEP_TIME, RETRY);
        zkClient = CuratorFrameworkFactory.builder()
                .retryPolicy(backoffRetry)
                .connectString(zookeeperAddress)
                .build();

        zkClient.start();

        try {
            if (!zkClient.blockUntilConnected(3000, TimeUnit.MILLISECONDS)) {
                throw new RuntimeException("cannot connect " + zookeeperAddress + "   zookeeper");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return zkClient;

    }


    public static void createServicePath(CuratorFramework zkClient, String servicePath) {

        try {
            if (REGISTERED_SERVICE_PATH_SET.contains(servicePath) || zkClient.checkExists().forPath(servicePath) != null) {
                log.info(servicePath + "  path  has exists");
            } else {
                zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(servicePath);
            }

            REGISTERED_SERVICE_PATH_SET.add(servicePath);

        } catch (Exception e) {
            throw new RuntimeException(e.getMessage());
        }


    }

    public static List<String> getChildrenNodes(CuratorFramework zkClient, String serviceName) {

        if (SERVICES_ADDRESS_MAP.containsKey(serviceName)) {
            return SERVICES_ADDRESS_MAP.get(serviceName);
        }

        String servicePath = ZRpcConstant.ZRPC_ROOT_PATH + serviceName;
        List<String> serviceList;
        try {
            serviceList = zkClient.getChildren().forPath(servicePath);
            SERVICES_ADDRESS_MAP.put(serviceName, serviceList);
            registerWatcher(zkClient, serviceName);
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage());
        }

        return serviceList;
    }

    private static void registerWatcher(CuratorFramework zkClient, String serviceName) {

        String servicePath = ZRpcConstant.ZRPC_ROOT_PATH + serviceName;

        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, servicePath, true);
        PathChildrenCacheListener listener = (((curatorFramework, pathChildrenCacheEvent) -> {
            List<String> serviceList = curatorFramework.getChildren().forPath(servicePath);
            SERVICES_ADDRESS_MAP.put(serviceName, serviceList);
            SERVICES_RECORD.updateServiceSet(serviceName,serviceList);
        }));

        pathChildrenCache.getListenable().addListener(listener);
        try {
            pathChildrenCache.start();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage());
        }

    }

    public static void deleteAllService(InetSocketAddress inetSocketAddress) {

        REGISTERED_SERVICE_PATH_SET.parallelStream().forEach(address -> {

            if (address.endsWith(inetSocketAddress.toString())) {
                try {
                    zkClient.delete().forPath(address);
                } catch (Exception e) {
                    throw new RuntimeException(e.getMessage());
                }
            }

        });

    }


}
